package MainLocale;

import java.util.ArrayList;
import java.util.Hashtable;

import worker.Worker;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.NodeException;

import Client.Client;
import Task.DataCheckInFile;
import Task.MapCompteMots;
import Task.ReduceCompteMots;
import Task.SplitCompteMots;
import Task.Task;

public class ClientLocale {

	/**
	 * Simulation en locale de la reception de workers distants sur le WorkFlow
	 * @param nbWorkers
	 * @return
	 * @throws ActiveObjectCreationException
	 * @throws NodeException
	 */
	public static ArrayList<Worker> getWorkers( int nbWorkers ) throws ActiveObjectCreationException, NodeException {
		
		ArrayList<Worker> workers = new ArrayList<Worker>();
		
		for(int i = 0 ; i < nbWorkers ; ++i ) {
			workers.add( new Worker() );
		}
		
		return workers;
	}
	
	/**
	 * Simulation en locale du Map/Reduce
	 * @param args
	 * @throws ActiveObjectCreationException
	 * @throws NodeException
	 * @throws ProActiveException
	 */
	public static void main (String[] args) throws ActiveObjectCreationException, NodeException , ProActiveException {
		
		//D�claration des variables.

		Task tache = new Task();
		Client client = new Client();
		ArrayList<Worker> workers = getWorkers( 3 );
		
		// Initialisation des variables.
		tache.setSplit( new SplitCompteMots() );
		tache.setMap( new MapCompteMots() );
		tache.setReduce( new ReduceCompteMots() );
		tache.setData( new DataCheckInFile() );
		client.setTache(tache);
			
			/*
			 * Recherche du Master sur le WorkFlow.
			 * 
			 * Worker workerContact = PAActiveObject.lookupActive(Worker.class, args[0] );
			 * workerContact.getManager.getMaster();
			 */
			
			//Initialisation de Data.
			client.getTache().getData().setData( client.readFromFile("./Data/test1.txt") );
			System.out.println("data : " + client.getTache().getData().getData() );
			
			// SPLIT()
			//							split(  					Data		, 		NbWorkers			).
			client.getTache().getSplit().split( client.getTache().getData().getData() , workers.size() );
			
			// pushData() + submit() + map()
			for(int i = 0; i < workers.size() ; ++i) {
				
				workers.get(i).pushData( tache.getSplit().getResult().get(i) );
				
				workers.get(i).submit( new MapCompteMots() ,  new ReduceCompteMots() , workers , i );
			
				Hashtable<String,Integer> htable = workers.get(i).getMap().map( workers.get(i).getData() );
				workers.get(i).getReduce().setHashtable( htable );
			}
			
			//	MAP()
			for(int i = 0; i < workers.size() ; ++i) {
				
				// Toute les tables de hashages sont synchronis�es.
				workers.get(i).getReduce().setHashtable( workers.get(i).getMap().map( workers.get(i).getData() ));
			}
			
			// Avant le reduce()
			for(int i = 0; i < workers.size() ; ++i) {
				
				System.out.println("avant (" + i + ") :" + workers.get(i).getReduce().getHashtable() );
			}
			
			//	REDUCE()
			for(int i = 0; i < workers.size() ; ++i) {
				
				// Toute les tables de hashages sont synchronis�es.
				workers.get(i).getReduce().reduce( workers , i );
			}
			
			// Apres le reduce()
			for(int i = 0; i < workers.size() ; ++i) {
				
				System.out.println("apres (" + i + ") :" + workers.get(i).getReduce().getHashtable() );
			}
			
			//Recup�ration des r�sultats : 
			for(int i = 0; i < workers.size() ; ++i) {
				
				client.addToResult( workers.get(i).getReduce().getHashtable() );
			}
			System.out.println("Resultat : " + client.getResult() );
			
	}
}
